package service;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import util.PropertiesUtils;

import java.util.Properties;

public class KafkaProducerService {

    public static void kafkaProducer(String value) {
        Properties props = new Properties();
        //broker地址
        props.put("bootstrap.servers", PropertiesUtils.getProperty("kafka_brokers"));
        //请求时候需要验证
        props.put("acks", "all");
        //请求失败时候需要重试
        props.put("retries", 0);
        //内存缓存区大小
        props.put("buffer.memory", 33554432);
        //指定消息key序列化方式
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        //指定消息本身的序列化方式
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>(PropertiesUtils.getProperty("kafka_target_topics"), value));
        System.out.println("Message sent successfully");
        producer.close();
    }
}
